package io.kiki.task.client.common.rpc.client;

import io.kiki.task.client.common.event.ChannelReconnectEvent;
import io.kiki.task.client.common.handler.ClientRegister;
import io.kiki.task.common.core.constant.SystemConstants.BEAT;
import io.kiki.task.common.core.context.SpringContext;
import io.kiki.task.common.core.model.HttpResponse;
import io.kiki.task.common.core.rpc.RpcContext;
import io.kiki.task.common.core.util.JsonUtil;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.TimeUnit;


@Slf4j
public class HttpClientHandler extends SimpleChannelInboundHandler<FullHttpResponse> {

    private final HttpClient httpClient;

    public HttpClientHandler(HttpClient httpClient) {
        this.httpClient = httpClient;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, FullHttpResponse fullHttpResponse) throws Exception {
        HttpHeaders httpHeaders = fullHttpResponse.headers();
        String content = fullHttpResponse.content().toString(CharsetUtil.UTF_8);
        if (log.isDebugEnabled()) {
            log.debug("Receive server data content:[{}], headers:[{}]", content, httpHeaders);
        }
        HttpResponse httpResponse = JsonUtil.parseObject(content, HttpResponse.class);
        RpcContext.invoke(httpResponse.getRequestId(), httpResponse, false);

    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        if (log.isDebugEnabled()) {
            log.debug("channelRegistered");
        }
        super.channelRegistered(ctx);
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        if (log.isDebugEnabled()) {
            log.debug("channelUnregistered");
        }
        ctx.channel().eventLoop().schedule(() -> {
            try {
                // 抛出重连事件
                SpringContext.getContext().publishEvent(new ChannelReconnectEvent());
                httpClient.reconnect();
            } catch (Exception e) {
                log.error("reconnect error ", e);
            }

        }, 10, TimeUnit.SECONDS);


    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        log.debug("channelActive");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        log.debug("channelInactive");
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        super.channelReadComplete(ctx);
        log.debug("channelReadComplete");
    }

    @Override
    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
        super.channelWritabilityChanged(ctx);
        log.debug("channelWritabilityChanged");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("kiki.task netty-http client exception", cause);
        super.exceptionCaught(ctx, cause);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        log.debug("userEventTriggered");
        if (evt instanceof IdleStateEvent) {
            ClientRegister.CLIENT.beat(BEAT.PING);
        } else {
            super.userEventTriggered(ctx, evt);
        }

    }
}
